package cn.rdtimes.wolfdsp.core.ha;

import cn.rdtimes.wolfdsp.core.invoker.http.ResponseMessage;
import cn.rdtimes.wolfdsp.core.service.ScheduleManager;
import cn.rdtimes.wolfdsp.core.util.StringUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.http.HttpStatus;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 缺省实现状态同步服务
 *
 * @author BZ
 */
public class DefaultStateSyncService implements StateSyncService {
    private static final Logger logger = LoggerFactory.getLogger(DefaultStateSyncService.class);
    /**
     * 缓存主服务器已调度的任务流程id，key-jobId，value-StateDTO
     */
    private final Map<String, StateDTO> stateMap;
    private final ExecutorService executor;

    private final String localIp;
    private final String masterIpAndPort;
    private final String standbyIpAndPort;
    private final CloseableHttpClient httpClient;
    private final RequestConfig requestConfig;

    public DefaultStateSyncService() {
        stateMap = new HashMap<>();
        localIp = getLocalIp();
        executor = Executors.newFixedThreadPool(2);
        masterIpAndPort = ScheduleManager.getInstance().getConfiguration().getMasterIpAndPort();
        standbyIpAndPort = ScheduleManager.getInstance().getConfiguration().getStandbyIpAndPort();
        httpClient = HttpClients.createDefault();
        requestConfig = RequestConfig.custom().setSocketTimeout(30_000).setConnectTimeout(10_000).
                setConnectionRequestTimeout(10_000).setMaxRedirects(3).build();
    }

    @Override
    public void shutdown() {
        try {
            executor.shutdown();
        } catch (Exception e) {
            //ignore
        }
        try {
            httpClient.close();
        } catch (Exception e) {
            // ignore
        }
    }

    @Override
    public void sendState(StateDTO stateDTO) throws IOException {
        if (ScheduleManager.getInstance().getPlatformHAStrategy().enabledPerform()
                && !StringUtil.isEmpty(localIp)) {
            executor.submit(new SendTask(stateDTO));
        } else {
            logger.warn("standby server can not send sync state");
        }
    }

    @Override
    public synchronized void receiveState(String json) throws Exception {
        if (ScheduleManager.getInstance().getPlatformHAStrategy().enabledPerform()) {
            logger.warn("master server can not receive sync state");
            throw new Exception("master server can not receive sync state");
        }

        StateDTO dto = JSON.parseObject(json, StateDTO.class);
        if (dto.getStateKind() == StateDTO.STATE_DEL) {
            stateMap.remove(dto.getJobId());
        } else if (dto.getStateKind() == StateDTO.STATE_ADD) {
            if (!stateMap.containsKey(dto.getJobId())) {
                stateMap.put(dto.getJobId(), dto);
            }
        }
    }

    @Override
    public synchronized Collection<StateDTO> getStateDTOList() {
        return stateMap.values();
    }

    private static String getLocalIp() {
        try {
            return InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            return "";
        }
    }


    private class SendTask implements Runnable {
        private final StateDTO dto;

        SendTask(StateDTO dto) {
            this.dto = dto;
        }

        @Override
        public void run() {
            String url = getStandbyUrl();
            if (StringUtil.isEmpty(url)) {
                logger.info("get standby ip and port is null");
                return;
            }

            JSONObject jsonObject = new JSONObject();
            jsonObject.put("stateKind", dto.getStateKind());
            jsonObject.put("jobId", dto.getJobId());
            try {
                HttpPost httpPost = new HttpPost(url);
                httpPost.setConfig(requestConfig);
                StringEntity entity = new StringEntity(jsonObject.toJSONString(), ContentType.APPLICATION_JSON);
                httpPost.setEntity(entity);
                CloseableHttpResponse response = httpClient.execute(httpPost);
                int statusCode = response.getStatusLine().getStatusCode();
                if (statusCode >= HttpStatus.SC_OK && statusCode < HttpStatus.SC_MULTIPLE_CHOICES) {
                    String respEntity = EntityUtils.toString(response.getEntity());
                    Map<String, Object> resp = (Map<String, Object>) JSON.parse(respEntity);
                    Object code = resp.get("code");
                    if (code == null || (int) code != ResponseMessage.SUCC) {
                        logger.error("sendState response error. code=" + code);
                    } else {
                        logger.debug("sendState is success");
                    }
                } else {
                    logger.error("sendState response error. statusCode=" + statusCode);
                }
            } catch (Exception e) {
                logger.error("sendState", e);
            }
        }

        private String getStandbyUrl() {
            String url = null;
            if (localIp.startsWith(masterIpAndPort)) {
                url = "http://" + standbyIpAndPort + "/" + StateSyncService.REQ_SYN_STATE_URL;
            } else if (localIp.startsWith(standbyIpAndPort)) {
                url = "http://" + masterIpAndPort + "/" + StateSyncService.REQ_SYN_STATE_URL;
            }
            return url;
        }

    }

}
